package com.qzy.flume;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;


public class MyInterceptor implements Interceptor{
    //1、获取配置文件的IP
    private String IP;
    private static final Logger logger = LoggerFactory
            .getLogger(MyInterceptor.class);

    public MyInterceptor(String ip) {
        this.IP = ip;
    }
    @Override
    public void initialize() {
        logger.info("*********MyInterceptor initialize.....**********");
    }

    @Override
    public Event intercept(Event event) {
        String body = new String(event.getBody(), Charset.forName("UTF-8"));
        try{
            // body为原始数据，newBody为处理后的数据
            logger.info("********拦截成功**********");
            logger.info("********body []**********"+body);

            String newBody = "[flume IP:"+IP+"]"+body ;
            logger.info("********newBody []**********"+newBody);
            event.setBody(newBody.toString().getBytes());
            logger.info("********event []**********"+event.toString());
        }catch (Exception e){
            e.printStackTrace();
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> intercepted = new ArrayList<>(events.size());
        for (Event event : events) {
            Event interceptedEvent = intercept(event);
            if (interceptedEvent != null) {
                intercepted.add(interceptedEvent);
            }
        }
        return intercepted;
    }

    @Override
    public void close() {
        logger.info("***************closing.....***********************");
    }

    public static class Builder implements Interceptor.Builder{
        //1、获取配置文件的appId
        private String IP;
        @Override
        public Interceptor build() {
            logger.info("***************build MyInterceptor .....***********************");
            return new MyInterceptor(IP);
        }

        @Override
        public void configure(Context context) {

        }
    }
}
